笔者|本文_《flask》flask+mqtt联动快速上手

篇首语：本文由编程笔记#小编为大家整理，主要介绍了《flask》flask+mqtt联动快速上手相关的知识，希望对你有一定的参考价值。

本文旨在介绍如何快速上手联动flask &＃43; mqtt&＃xff0c;本文将会给出一个简单的demo&＃xff0c;用于演示在如何通过访问flask接口来触发mqtt&＃xff0c;并在flask运行的基础的上对mqtt进行订阅。

因为有项目需求&＃xff0c;所以需要flask &＃43; mqtt进行联动&＃xff0c;因为需要一直开启监听&＃xff0c;所以需要一直挂在一个线程上&＃xff0c;一开始想到用多线程做&＃xff0c;或者说用异步协程来做&＃xff0c;后来发现有一个关于flask的mqtt扩展库&＃xff0c;因此为了快速上手&＃xff0c;直接用该库进行开发&＃xff0c;可以节省很多精力。

link: https://github.com/stlehmann/Flask-MQTT

首先&＃xff0c;我们需要安装flask-mqtt库。

pip install flask-mqtt


下面放上一个简单的示例代码&＃xff0c;里面包括了MQTT的订阅、发布&＃xff0c;flask的访问。

"""
A small Test application to show how to use Flask-MQTT.
"""

import eventlet
from flask import Flask, render_template
from flask_mqtt import Mqtt


eventlet.monkey_patch()

app &＃61; Flask(__name__)
app.config[&＃39;SECRET&＃39;] &＃61; &＃39;my secret key&＃39;
app.config[&＃39;TEMPLATES_AUTO_RELOAD&＃39;] &＃61; True
app.config[&＃39;MQTT_BROKER_URL&＃39;] &＃61; &＃39;broker.emqx.io&＃39;
app.config[&＃39;MQTT_BROKER_PORT&＃39;] &＃61; 1883
app.config[&＃39;MQTT_USERNAME&＃39;] &＃61; &＃39;&＃39;
app.config[&＃39;MQTT_PASSWORD&＃39;] &＃61; &＃39;&＃39;
app.config[&＃39;MQTT_KEEPALIVE&＃39;] &＃61; 5
app.config[&＃39;MQTT_TLS_ENABLED&＃39;] &＃61; False
app.config[&＃39;MQTT_CLEAN_SESSION&＃39;] &＃61; True


mqtt &＃61; Mqtt(app)


&＃64;app.route(&＃39;/&＃39;)
def index():
return "hello mqtt_flask"

&＃64;app.route(&＃39;/hello&＃39;)
def hello():
mqtt.publish(&＃39;hello&＃39;, &＃39;hello, this is flask&＃39;)
print("[mqtt] publish successfully")
return "publish successfully"

&＃64;mqtt.on_connect()
def handle_connect(client, userdata, flags, rc):
mqtt.subscribe(&＃39;hello&＃39;)
print("[mqtt] has listen topic hello")


&＃64;mqtt.on_message()
def handle_mqtt_message(client, userdata, message):
data &＃61; dict(
topic&＃61;message.topic,
payload&＃61;message.payload.decode()
)
print(data)

if __name__ &＃61;&＃61; &＃39;__main__&＃39;:
# important: Do not use reloader because this will create two Flask instances.
# Flask-MQTT only supports running with one instance # socketio.run(app, host&＃61;&＃39;0.0.0.0&＃39;, port&＃61;5000, use_reloader&＃61;False, debug&＃61;False) app.run()


订阅测试

这里用了MQTTX工具来进行测试&＃xff0c;现在我已经连接到了一个MQTT服务器上&＃xff0c;我将用MQTTX向该服务器发送一个topic为hello&＃xff0c;payload为"msg": "Hello, This is mqtt server"的消息。

运行程序后&＃xff0c;MQTTX发送消息&＃xff0c;结果如下&＃xff0c;可以看到&＃xff0c;接收正常。

发布测试

现在我们访问http://localhost:5000/hello&＃xff0c;flask服务器将会向我的MQTT服务器发送一个topic为hello&＃xff0c;payload为"hello, this is flask"的消息&＃xff0c;我们用MQTTX来监听。

从下图中可以看到&＃xff0c;MQTTX成功订阅接收到了该消息&＃xff0c;至此&＃xff0c;关于FLASK&＃43;MQTT的必要流程已经可以走通了。

最初&＃xff0c;笔者以为启动mqtt_client.loop_start()函数之后会线程堵塞&＃xff0c;但是在使用 Paho MQTT Python 库时&＃xff0c;调用 client.loop_start() 方法会在后台启动一个新线程&＃xff0c;用于处理 MQTT 客户端的事件循环。这意味着您的代码不会堵塞&＃xff0c;并且可以在 MQTT 客户端的事件循环运行时继续执行。

例如&＃xff0c;您可以在调用 client.loop_start() 后使用以下代码继续执行其他操作&＃xff1a;

import paho.mqtt.client as mqtt
# The callback for when a message is received from the server
def on_message(client, userdata, message):
print(f"Received message &＃39;message.payload.decode()&＃39; on topic &＃39;message.topic&＃39;")
# Create an MQTT client
client &＃61; mqtt.Client()
# Set the on_message callback
client.on_message &＃61; on_message
# Connect to the MQTT server
client.connect("mqtt.example.com", 1883, 60)
# Subscribe to a topic
client.subscribe("my/topic")
# Start the MQTT client loop in the background
client.loop_start()
# Do other work here
print("Doing other work")
# Wait for a while
time.sleep(10)
# Stop the MQTT client loop
client.loop_stop()


在上面的代码中&＃xff0c;我们在调用 client.loop_start() 后立即执行了其他代码&＃xff0c;然后等待了 10 秒钟。在这期间&＃xff0c;MQTT 客户端的事件循环仍在后台运行&＃xff0c;并等待接收消息。

基于此&＃xff0c;那么flask和mqtt联动其实不用上述的扩展库也可以&＃xff0c;笔者在paho官方的demo上小改一下&＃xff0c;最后结果如下所示&＃xff1a;

"""
访问 localhost:5000/hello时&＃xff0c;会用mqtt客户端发布主题消息
"""
from flask import Flask, jsonify
import paho.mqtt.client as mqtt
import sys


app &＃61; Flask(__name__)
print(&＃39;[app] start work&＃39;, file&＃61;sys.stdout)


def connect_mqtt() -> mqtt:
def on_connect(client, userdata, flags, rc):
if rc &＃61;&＃61; 0:
print("Connected to MQTT Broker!", file&＃61;sys.stdout)
else:
print("Failed to connect, return code %d\\n", rc, file&＃61;sys.stdout)

client &＃61; mqtt.Client()
client.username_pw_set("", "")
client.on_connect &＃61; on_connect
client.connect("broker.emqx.io", 1883, 60)
return client


def subscribe(client: mqtt, topic):
def on_message(client, userdata, msg):
print(f"[mqtt] Received &＃96;msg.payload.decode()&＃96; from &＃96;msg.topic&＃96; topic", file&＃61;sys.stdout)

client.on_message &＃61; on_message
client.subscribe(topic)
print("[mqtt] subscribe topic")


client &＃61; connect_mqtt()
subscribe(client, "hello")
client.loop_start()

&＃64;app.route("/hello")
def alarm():
client.publish("hello", "important msg")
print("[mqtt] send msg successfully", file&＃61;sys.stdout)
return "Mqtt message published"


if __name__ &＃61;&＃61; &＃39;__main__&＃39;:
app.run()


该代码经本人实际测试&＃xff0c;也是可以正常食用的。需要说明的是&＃xff0c;如果你只是需要发布而不需要监听&＃xff0c;那么client.loop_start()不是必要的。